package com.anli02.RabbitMQConfig;

import com.anli02.entity.pojo.OnlineMusic;
import com.anli02.entity.vo.OnlineMusicDoc;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

import java.io.DataInput;
import java.util.HashSet;
import java.util.Set;

@Component
public  class MusicConsumer
{
@Autowired
    RestHighLevelClient client;
    // 模拟存储已处理的 MessageId，实际可以用 Redis 或数据库替代
    private static final Set<String> processedMessageIds = new HashSet<>();
    @RabbitListener(queues="direct.music")
    public void listen(Message message)
    {
        // 模拟存储已处理的 MessageId，实际可以用 Redis 或数据库替代

        String messageId = message.getMessageProperties().getMessageId();

        // 检查消息是否已处理
        if (messageId != null && processedMessageIds.contains(messageId)) {
            System.out.println("重复消息，已跳过处理：" + messageId);
            return; // 消息已处理，跳过
        }
        OnlineMusic onlineMusic=new OnlineMusic();
        //同步到es
        // 将接收到的消息转换为OnlineMusic对象
        byte[] body = message.getBody();
        String jsonString = new String(body);
        try {
             onlineMusic = new ObjectMapper().readValue( jsonString, OnlineMusic.class);
        }catch (Exception e)
        {
            e.printStackTrace();
        }

        OnlineMusicDoc onlineMusicDoc=new OnlineMusicDoc(onlineMusic);
        IndexRequest request = new IndexRequest("music");
        request.id(onlineMusicDoc.getMusic()); // 设置文档ID
        ///指定数据的格式
        request.source(onlineMusicDoc, XContentType.JSON);
        ///2默认
        try {
            client.index(request,RequestOptions.DEFAULT);
        }catch (Exception e)
        {
            throw  new MessagingException("异常" );

        }
        // 处理完成后，记录 MessageId
        processedMessageIds.add(messageId);


    }

}